package main

import (
	"context"
	"database/sql"
	dbsql "database/sql"
	"encoding/json"
	"fmt"
	"log"
	"mig2kwdb/common"
	"mig2kwdb/controller"
	"mig2kwdb/readers"
	"mig2kwdb/writers"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgproto3"
	_ "github.com/taosdata/driver-go/v3/taosWS"
)

var mysqlConnInfo common.DBConnectionInfo
var postgresqlConnInfo common.DBConnectionInfo
var kwdbConnInfo common.DBConnectionInfo
var sqlserverConnInfo common.DBConnectionInfo
var oracleConnInfo common.DBConnectionInfo
var tdengineConnInfo common.DBConnectionInfo
var dmConnInfo common.DBConnectionInfo

func init() {
	//mysql
	mysqlConnInfo.Dbtype = "mysql"
	mysqlConnInfo.Ip = "10.20.50.63"
	mysqlConnInfo.Port = 3306
	mysqlConnInfo.User = "root"
	mysqlConnInfo.Password = "Easync.2019"
	//postgresql
	postgresqlConnInfo.Dbtype = "postgresql"
	postgresqlConnInfo.Ip = "10.20.50.115"
	postgresqlConnInfo.Port = 5432
	postgresqlConnInfo.User = "postgres"
	postgresqlConnInfo.Password = "easync.2019"
	postgresqlConnInfo.DatabaseName = "postgres"
	//kwdb
	kwdbConnInfo.Dbtype = "kwdb"
	kwdbConnInfo.Ip = "10.20.50.93"
	kwdbConnInfo.Port = 26257
	kwdbConnInfo.User = "root"
	kwdbConnInfo.Password = ""
	//sql server
	sqlserverConnInfo.Dbtype = "sqlserver"
	sqlserverConnInfo.Ip = "10.20.50.115"
	sqlserverConnInfo.Port = 1433
	sqlserverConnInfo.User = "sa"
	sqlserverConnInfo.Password = "Ocean@2025"
	//oracle
	oracleConnInfo.Dbtype = "oracle"
	oracleConnInfo.Ip = "10.20.50.79"
	oracleConnInfo.Port = 1521
	oracleConnInfo.User = "C##ERDSUSER"
	oracleConnInfo.Password = "Ocean@2025"
	oracleConnInfo.InstanceName = "orarac"
	oracleConnInfo.Method = 2 //1:sid,2:service name

	//tdengine
	tdengineConnInfo.Dbtype = "tdengine"
	tdengineConnInfo.Ip = "10.20.50.115"
	tdengineConnInfo.Port = 6041
	tdengineConnInfo.User = "root"
	tdengineConnInfo.Password = "taosdata"

	//dameng
	dmConnInfo.Dbtype = "dameng"
	dmConnInfo.Ip = "10.20.50.115"
	dmConnInfo.Port = 5236
	dmConnInfo.User = "SYSDBA"
	dmConnInfo.Password = "Ocean@2025"
}
func Test_dameng(t *testing.T) {
	dameng, err := readers.CreateReader(dmConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	defer dameng.Close()
	dbs, err := dameng.GetDataBases()
	if err != nil {
		log.Println(err)
		return
	}
	log.Println(dbs)
	tables, err := dameng.GetDatabaseTables("DMHR", "")
	if err != nil {
		log.Println(err)
		return
	}
	log.Println(tables)
	def, err := dameng.GetTableDefinition("DMHR", "", "ALL_TYPES")
	if err != nil {
		log.Println(err)
		return
	}
	rows, err := dameng.GetConditionRows("DMHR", "", "ALL_TYPES", def.Columns, "")
	if err != nil {
		log.Println(err)
		return
	}
	defer rows.Close()

	for rows.Next() {
		value := make([]interface{}, len(def.Columns))
		for i := 0; i < len(def.Columns); i++ {
			value[i] = &sql.NullString{}
		}
		err = rows.Scan(value...)
		if err != nil {
			log.Println(err)
			return
		}
		log.Println(value)
	}

	log.Println(def)

}
func Test_kwdbselect(t *testing.T) {
	
	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	defer kwdb.Close()
	sql := "select * from oracle_db.public.test"
	rows, _ := kwdb.GetDB().Query(sql)
	columns, _ := rows.Columns()
	sourceValue := make([]interface{}, len(columns))
	for i := 0; i < len(columns); i++ {
		sourceValue[i] = &dbsql.NullString{}
	}
	for rows.Next() {

		rows.Scan(sourceValue...)

	}

}
func Test_KwdbWriter(t *testing.T) {

	dbName := "test001"
	tableName := "test1"
	reader, err := readers.CreateReader(mysqlConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	defer reader.Close()
	def, err := reader.GetTableDefinition(dbName, "", tableName)
	if err != nil {
		log.Fatal(err)
	}

	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	defer kwdb.Close()

	kwdb_schemaName := "public"
	kwdb_dbName := "testdb"
	kwdb_tableName := tableName
	err = kwdb.DropTable(kwdb_dbName, kwdb_schemaName, kwdb_tableName, false)
	err = kwdb.CreateTable(kwdb_dbName, kwdb_schemaName, kwdb_tableName, def, false)

	log.Println(err)

	//
	sql_str := "select * from `" + dbName + "`.`" + tableName + "`"
	rows, err := reader.GetDB().Query(sql_str)
	if err != nil {
		log.Println(err)
		return
	}
	defer rows.Close()
	var values [][]interface{}
	num := 0
	for rows.Next() {
		value := make([]interface{}, len(def.Columns))
		for i := 0; i < len(def.Columns); i++ {
			value[i] = &sql.NullString{}
		}
		err := rows.Scan(value...)
		if err != nil {
			log.Println(err)
			return
		}
		values = append(values, value)
		if num > 10000 {

			kwdb.InsertRecords(kwdb_dbName, kwdb_schemaName, kwdb_tableName, def.Columns, values, "sync", false)
			num = 0
			values = values[0:0]
		}
		num++
	}
	if len(values) > 0 {
		kwdb.InsertRecords(kwdb_dbName, kwdb_schemaName, kwdb_tableName, def.Columns, values, "sync", false)
	}
}
func Test_kwdb_dameng(t *testing.T) {
	reader, err := readers.CreateReader(dmConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	defer reader.GetDB().Close()
	columns, err := reader.GetTableColumn("DMHR", "", "ALL_TYPES")
	if err != nil {
		log.Println(err)
		return
	}
	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	defer kwdb.Close()
	rowsSource, err := kwdb.GetConditionRows("dameng_database", "public", "ALL_TYPES", columns, "")
	if err != nil {
		log.Println(err)
		return
	}
	defer rowsSource.Close()
	sourceValue := make([]interface{}, len(columns))
	for i := 0; i < len(columns); i++ {

		sourceValue[i] = &sql.NullString{}

	}
	var num int64 = 0
	for rowsSource.Next() {
		err = rowsSource.Scan(sourceValue...)
		if err != nil {
			log.Println(err)
			continue
		}
		num++
	}
	log.Println(num)
}
func Test_kwdb_rows(t *testing.T) {
	reader, err := readers.CreateReader(tdengineConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	defer reader.GetDB().Close()
	columns, err := reader.GetTableColumn("test", "", "meters")
	if err != nil {
		log.Println(err)
		return
	}
	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	defer kwdb.Close()
	rowsSource, err := kwdb.GetConditionRows("tdengine_db", "public", "meters", columns, "")
	if err != nil {
		log.Println(err)
		return
	}
	defer rowsSource.Close()
	sourceValue := make([]interface{}, len(columns))
	for i := 0; i < len(columns); i++ {
		if strings.ToUpper(columns[i].Type) == "TIMESTAMP" {
			sourceValue[i] = &time.Time{}
		} else {
			sourceValue[i] = &sql.NullString{}
		}
	}
	var num int64 = 0
	for rowsSource.Next() {
		err = rowsSource.Scan(sourceValue...)
		if err != nil {
			log.Println(err)
			continue
		}
		num++
	}
	log.Println(num)
}
func Test_kwdb_connect(t *testing.T) {
	log.Println("Test_connect")
	var connInfo common.DBConnectionInfo
	connInfo.Dbtype = "kwdb"
	connInfo.Ip = "10.20.50.93"
	connInfo.Port = 26257
	connInfo.User = "root"
	connInfo.Password = ""
	kwdb, err := writers.CreateWriter(connInfo)
	db := kwdb.GetDB()
	log.Println(db, err)
	if err != nil {
		log.Println(err)
		return
	}
	defer db.Close()
	log.Println(db.Ping())
	sql := `INSERT INTO sensor_data.readings 
	VALUES 
	(NOW(), 23.0, 59.5, 101, 'Room 101'),
	(NOW(), 23.5, 58.9, 102, 'Room 102'),
	(NOW(), 19.8, 65.5, 103, 'Room 103');`
	res, err := db.Exec(sql)
	if err != nil {
		kwdb.IsNetworkError(err)
		log.Println(err)
		return
	}
	log.Println(res, err)

}
func Test_postrep(t *testing.T) {
	dsn := "postgres://" + postgresqlConnInfo.User + ":" + postgresqlConnInfo.Password + "@" + postgresqlConnInfo.Ip + ":" + fmt.Sprint(postgresqlConnInfo.Port) + "/postgres?replication=database"

	const outputPlugin = "pgoutput"
	//const outputPlugin = "wal2json"
	conn, err := pgconn.Connect(context.Background(), dsn)
	if err != nil {
		log.Fatalln("failed to connect to PostgreSQL server:", err)
	}
	defer conn.Close(context.Background())

	result := conn.Exec(context.Background(), "DROP PUBLICATION IF EXISTS pglogrepl_demo;")
	_, err = result.ReadAll()
	if err != nil {
		log.Fatalln("drop publication if exists error", err)
	}

	result = conn.Exec(context.Background(), "CREATE PUBLICATION pglogrepl_demo FOR ALL TABLES;")
	_, err = result.ReadAll()
	if err != nil {
		log.Fatalln("create publication error", err)
	}
	log.Println("create publication pglogrepl_demo")

	var pluginArguments []string
	var v2 bool
	if outputPlugin == "pgoutput" {
		// streaming of large transactions is available since PG 14 (protocol version 2)
		// we also need to set 'streaming' to 'true'
		pluginArguments = []string{
			"proto_version '1'",
			"publication_names 'pglogrepl_demo'",
		}
		v2 = true
		// uncomment for v1
		// pluginArguments = []string{
		//	"proto_version '1'",
		//	"publication_names 'pglogrepl_demo'",
		//	"messages 'true'",
		// }
	} else if outputPlugin == "wal2json" {
		pluginArguments = []string{"\"pretty-print\" 'true'"}
	}
	log.Println(v2)
	sysident, err := pglogrepl.IdentifySystem(context.Background(), conn)
	if err != nil {
		log.Fatalln("IdentifySystem failed:", err)
	}
	log.Println("SystemID:", sysident.SystemID, "Timeline:", sysident.Timeline, "XLogPos:", sysident.XLogPos, "DBName:", sysident.DBName)

	slotName := "pglogrepl_demo"

	_, err = pglogrepl.CreateReplicationSlot(context.Background(), conn, slotName, outputPlugin, pglogrepl.CreateReplicationSlotOptions{Temporary: true})
	if err != nil {
		log.Fatalln("CreateReplicationSlot failed:", err)
	}
	log.Println("Created temporary replication slot:", slotName)

	err = pglogrepl.StartReplication(context.Background(), conn, slotName, sysident.XLogPos, pglogrepl.StartReplicationOptions{PluginArgs: pluginArguments})
	if err != nil {
		log.Fatalln("StartReplication failed:", err)
	}
	log.Println("Logical replication started on slot", slotName)

	clientXLogPos := sysident.XLogPos
	standbyMessageTimeout := time.Second * 10
	nextStandbyMessageDeadline := time.Now().Add(standbyMessageTimeout)

	for {
		if time.Now().After(nextStandbyMessageDeadline) {
			err = pglogrepl.SendStandbyStatusUpdate(context.Background(), conn, pglogrepl.StandbyStatusUpdate{WALWritePosition: clientXLogPos})
			if err != nil {
				log.Fatalln("SendStandbyStatusUpdate failed:", err)
			}
			log.Printf("Sent Standby status message at %s\n", clientXLogPos.String())
			nextStandbyMessageDeadline = time.Now().Add(standbyMessageTimeout)
		}

		ctx, cancel := context.WithDeadline(context.Background(), nextStandbyMessageDeadline)
		rawMsg, err := conn.ReceiveMessage(ctx)
		cancel()
		if err != nil {
			if pgconn.Timeout(err) {
				continue
			}
			log.Fatalln("ReceiveMessage failed:", err)
		}

		if errMsg, ok := rawMsg.(*pgproto3.ErrorResponse); ok {
			log.Fatalf("received Postgres WAL error: %+v", errMsg)
		}

		msg, ok := rawMsg.(*pgproto3.CopyData)
		if !ok {
			log.Printf("Received unexpected message: %T\n", rawMsg)
			continue
		}

		switch msg.Data[0] {
		case pglogrepl.PrimaryKeepaliveMessageByteID:
			pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
			if err != nil {
				log.Fatalln("ParsePrimaryKeepaliveMessage failed:", err)
			}
			log.Println("Primary Keepalive Message =>", "ServerWALEnd:", pkm.ServerWALEnd, "ServerTime:", pkm.ServerTime, "ReplyRequested:", pkm.ReplyRequested)
			if pkm.ServerWALEnd > clientXLogPos {
				clientXLogPos = pkm.ServerWALEnd
			}
			if pkm.ReplyRequested {
				nextStandbyMessageDeadline = time.Time{}
			}

		case pglogrepl.XLogDataByteID:
			xld, err := pglogrepl.ParseXLogData(msg.Data[1:])
			if err != nil {
				log.Fatalln("ParseXLogData failed:", err)
			}

			if outputPlugin == "wal2json" {
				log.Printf("wal2json data: %s\n", string(xld.WALData))
			} else {
				log.Printf("XLogData => WALStart %s ServerWALEnd %s ServerTime %s WALData:\n", xld.WALStart, xld.ServerWALEnd, xld.ServerTime)

			}

			if xld.WALStart > clientXLogPos {
				clientXLogPos = xld.WALStart
			}
		}
	}
}
func Test_tdenginews(t *testing.T) {
	// 构造 DSN，使用 WebSocket 协议，假设 taosAdapter 运行在本地 6041 端口
	var dsn = "root:taosdata@ws(10.20.50.115:6041)/test" // 连接到默认数据库
	// 如需指定数据库，可使用： "root:taosdata@ws(localhost:6041)/test"

	// 打开数据库连接
	db, err := sql.Open("taosWS", dsn)
	if err != nil {
		log.Fatalf("Failed to open connection: %v", err)
	}
	defer db.Close()

	// 尝试连接
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := db.PingContext(ctx); err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	fmt.Println("Connected to TDengine via WebSocket!")

	properties := "group.id=test_consumer|" +
		"auto.offset.reset=earliest|" + // 从最早开始消费，可选 latest
		"enable.auto.commit=false|" + // 建议手动提交偏移量
		"msg.with.table.name=true" // 在消息中包含表名
	// 示例：执行一个简单查询
	rows, err := db.Query("SELECT * from topic_test_device_data", properties)
	if err != nil {
		log.Fatalf("Query failed: %v", err)
	}
	defer rows.Close()
	var num int64 = 0

	columns, _ := rows.Columns()
	for rows.Next() {
		value := make([]interface{}, len(columns))
		for i := 0; i < len(columns); i++ {
			value[i] = &sql.NullString{}
		}
		err := rows.Scan(value...)
		if err != nil {
			log.Println(err)
			return
		}
		if num%10000 == 0 {
			log.Println("Processing row:", num)
		}
		num++

	}
	log.Println("Total rows:", num)
}
func Test_tdengine(t *testing.T) {
	reader, err := readers.CreateReader(tdengineConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	err = reader.GetDB().Ping()
	log.Println(err)
	dbs, err := reader.GetDataBases()
	log.Println(dbs, err)
	def, _ := reader.GetTableDefinition("test", "", "meters")
	log.Println(def)
	for _, db := range dbs {
		tables, err := reader.GetDatabaseTables(db, "")
		log.Println(tables, err)
		for _, table := range tables {
			columns, err := reader.GetTableColumn(db, "", table)
			log.Println(columns, err)
			rows, err := reader.GetDB().Query("select ts from " + db + "." + table)
			var num int64 = 0
			if err == nil {
				for rows.Next() {
					var ts sql.NullString
					rows.Scan(&ts)
					num++
				}
			}
			log.Println(num, err)
		}
	}
}
func Test_tdengine_rows(t *testing.T) {
	reader, err := readers.CreateReader(tdengineConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	defer reader.GetDB().Close()
	columns, err := reader.GetTableColumn("test", "", "meters")
	if err != nil {
		log.Println(err)
		return
	}
	sourceValue := make([]interface{}, len(columns))
	rowsSource, err := reader.GetConditionRows("test", "", "meters", columns, "")
	if err != nil {
		log.Println(err)
		return
	}
	defer rowsSource.Close()
	for i := 0; i < len(columns); i++ {
		if strings.ToUpper(columns[i].Type) == "TIMESTAMP" {
			sourceValue[i] = &time.Time{}
		} else {
			sourceValue[i] = &sql.NullString{}
		}
	}
	var num int64 = 0
	for rowsSource.Next() {
		err = rowsSource.Scan(sourceValue...)
		if err != nil {
			log.Println(err)
			continue
		}
		num++
	}
	log.Println(num)
}
func Test_postgresql(t *testing.T) {
	reader, err := readers.CreateReader(postgresqlConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	err = reader.GetDB().Ping()
	log.Println(err)
	dbs, err := reader.GetDataBases()
	log.Println(dbs, err)
	def, _ := reader.GetTableDefinition("postgres", "public", "testall_3")
	log.Println(def)
}
func Test_sqlserver(t *testing.T) {
	reader, err := readers.CreateReader(sqlserverConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	err = reader.GetDB().Ping()
	log.Println(err)
	dbs, err := reader.GetDataBases()
	log.Println(dbs, err)
	def, _ := reader.GetTableDefinition("test", "dbo", "Table_1")
	log.Println(def)
}
func Test_oracle(t *testing.T) {
	reader, err := readers.CreateReader(oracleConnInfo)
	if err != nil {
		log.Fatal(err)
	}
	err = reader.GetDB().Ping()
	log.Println(err)
	dbs, err := reader.GetDataBases()
	log.Println(dbs, err)
	for _, db := range dbs {
		tables, err := reader.GetDatabaseTables(db, "")
		log.Println(tables, err)
	}
	def, _ := reader.GetTableDefinition("C##ERDSUSER", "", "COUNTRIES")
	log.Println(def)
}
func Test_tdengine2kwdb(t *testing.T) {
	tdenginedb, err := readers.CreateReader(tdengineConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	c := controller.NewController(tdenginedb, kwdb)
	var stats common.TaskStattistics
	var sourceDB, targetDB common.DatabaseConfig
	sourceDB.Name = "test"
	targetDB.Name = "tdengine_db"
	targetDB.Schema = "public"
	err = c.SyncOneDB(ctx, &stats, sourceDB, targetDB, true)
	log.Println(err)
}
func Test_postgresql2kwdb(t *testing.T) {
	postgresdb, err := readers.CreateReader(postgresqlConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	c := controller.NewController(postgresdb, kwdb)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	var stats common.TaskStattistics
	var sourceDB, targetDB common.DatabaseConfig
	sourceDB.Name = "postgresql"
	sourceDB.Schema = "public"
	targetDB.Name = "postgresql_db"
	targetDB.Schema = "public"
	err = c.SyncOneDB(ctx, &stats, sourceDB, targetDB, false)
	log.Println(err)
}
func Test_oracle2kwdb(t *testing.T) {
	oracledb, err := readers.CreateReader(oracleConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	c := controller.NewController(oracledb, kwdb)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	var stats common.TaskStattistics
	var sourceDB, targetDB common.DatabaseConfig
	sourceDB.Name = "C##ERDSUSER"
	targetDB.Name = "oracle_users"
	targetDB.Schema = "public"
	err = c.SyncOneDB(ctx, &stats, sourceDB, targetDB, false)
	log.Println(err)
}
func Test_mysql2kwdb(t *testing.T) {
	mysqldb, err := readers.CreateReader(mysqlConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	kwdb, err := writers.CreateWriter(kwdbConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	c := controller.NewController(mysqldb, kwdb)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	var stats common.TaskStattistics
	var sourceDB, targetDB common.DatabaseConfig
	sourceDB.Name = "tpcc"
	targetDB.Name = "tpcc_mysql"
	targetDB.Schema = "public"
	err = c.SyncOneDB(ctx, &stats, sourceDB, targetDB, false)
	log.Println(err)
}
func Test_mysql(t *testing.T) {
	log.Println("Test_connect")
	reader, err := readers.CreateReader(mysqlConnInfo)
	if err != nil {
		log.Println(err)
		return
	}
	defer reader.Close()
	//get all databases
	pos, err := reader.GetStartPos()
	log.Println(pos, err)
	databases, err := reader.GetDataBases()
	if err != nil {
		reader.IsNetworkError(err)
		return
	}
	log.Println(err, databases)
	//get the tables of tpcc
	tables, err := reader.GetDatabaseTables("tpcc", "")
	log.Println(err, tables)
	def, _ := reader.GetTableDefinition("tpcc", "", "warehouse")
	log.Println(def)
	def, _ = reader.GetTableDefinition("testdatabase", "", "new_table")
	log.Println(def)
	def, _ = reader.GetTableDefinition("testdatabase", "", "new_table2")
	log.Println(def)
}

func Test_synctask(t *testing.T) {
	syncTaskJson := `
	{
		"name": "database_sync_task",
		"description": "数据库同步任务示例",
		"source": {
		  "dbcon": {
			"Ip": "192.168.1.100",
			"Port": 3306,
			"User": "source_user",
			"Password": "source_password",
			"Dbtype": "mysql",
			"SSL": false,
			"Compress": false,
			"Method": 0,
			"DatabaseName": "source_db",
			"InstanceName": ""
		  },
		  "databases": [
			{
			  "name": "source_database_1",
			  "schema": "public",
			  "include": [
				"users",
				"orders",
				"products"
			  ],
			  "exclude": [
				"logs",
				"temporary_data"
			  ]
			},
			{
			  "name": "source_database_2",
			  "schema": "business",
			  "include": [],
			  "exclude": [
				"audit_trail"
			  ]
			}
		  ]
		},
		"target": {
		  "dbcon": {
			"Ip": "192.168.1.200",
			"Port": 26257,
			"User": "target_user",
			"Password": "target_password",
			"Dbtype": "kwdb",
			"SSL": false,
			"Compress": false,
			"Method": 0,
			"DatabaseName": "target_db",
			"InstanceName": ""
		  },
		  "databases": [
			{
			  "name": "target_database",
			  "schema": "public",
			  "include": [
				"users",
				"orders",
				"products"
			  ],
			  "exclude": []
			}
		  ]
		}
	  }`
	var syncTask common.MigrateTask
	json.Unmarshal([]byte(syncTaskJson), &syncTask)
	log.Println(syncTask)
}
